package org.sg20.labelpropagation.lpaiter;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.sg20.utils.LpaDotVal;


import java.io.IOException;

public class LpaIterReducer extends Reducer<Text, Text, Text, Text> {

    private int labelNum; //标签总数
    private int propagateNum; //传播的标签数量
    private Text outKey;
    private Text outVal;

    @Override
    protected void setup(Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        super.setup(context);
        outKey=new Text();
        outVal=new Text();
        this.labelNum=context.getConfiguration().getInt("label-num",1);
        this.propagateNum=context.getConfiguration().getInt("propagate-num",1);
    }

    @Override
    protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        String links = null;
        int label;
        double prob;
        String [] label_prob;
        LpaDotVal dotVal=new LpaDotVal(labelNum);

        for(Text value:values){
            String strValue=value.toString();
            if(strValue.startsWith("[")){
                //是链接列表
                links=strValue;
            }else{
                //求该点各个标签的取值概率
                label_prob=strValue.split(",");
                label=Integer.parseInt(label_prob[0]);
                prob=Double.parseDouble(label_prob[1]);
                dotVal.addProb(label,prob);
            }
        }
        //对标签概率向量进行归一化
        int[] labels=dotVal.getLabel(propagateNum);
        dotVal.normalization(labels);
        //输出标签概率向量和链接关系
        outKey=key;
        outVal.set(dotVal +" "+links);
        context.write(outKey,outVal);
    }
}
